package com.xiaoyilin;

import org.reactivestreams.Subscription;
import reactor.core.publisher.BaseSubscriber;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;

import java.util.Arrays;
import java.util.Optional;

/**
 * 订阅并触发数据流
 * subscribe();
 * 订阅并指定对正常数据元素如何处理
 * subscribe(Consumer<? super T> consumer);
 * 订阅并定义对正常数据元素和错误信号的处理
 * subscribe(Consumer<? super T> consumer,Consumer<? super Throwable> errorConsumer);
 * 订阅并定义对正常数据元素、错误信号和完成信号的处理
 * subscribe(Consumer<? super T> consumer,Consumer<? super Throwable> errorConsumer,Runnable completeConsumer);
 * 订阅并定义对正常数据元素、错误信号和完成信号的处理，以及订阅发生时的处理逻辑
 * subscribe(Consumer<? super T> consumer,Consumer<? super Throwable> errorConsumer,Runnable completeConsumer,Consumer<? super Subscription> subscriptionConsumer);
 */
public class ReactorTest {

    public static void main(String[] args) {
        Flux.just("tom", "jack", "allen").flatMap(s -> Flux.fromArray(s.split("\\s*"))).map(s -> s.concat("@qq.com")).subscribe(System.out::println);
        Flux.fromIterable(Arrays.asList("foo", "bar", "foobar"));
        Integer[] array = new Integer[]{1, 2, 3, 4, 5, 6};
        Flux.fromArray(array);
        Flux.fromStream(Arrays.asList(array).stream());
        // 只有完成信号的空数据流
        Mono.empty();
        Flux.just();
        Flux.empty();
        Mono.justOrEmpty(Optional.empty());
        // 只有错误信号的数据流
        Flux.error(new Exception("some error"));
        Mono.error(new Exception("some error"));
        Flux<Integer> numbersFromFiveToSeven = Flux.range(5, 3);
        numbersFromFiveToSeven.subscribe(System.out::println, System.err::println,
                () -> System.out.println("Completed"));

        Mono.error(new Exception("some error")).subscribe(
                System.out::println,
                System.err::println,
                () -> System.out.println("Completed!")
        );


        Flux.range(1, 10)
                .filter(i -> i % 2 == 1)
                .doOnRequest(r -> System.out.println("request of " + r))
                .subscribe(new BaseSubscriber<Integer>() {

                    @Override
                    protected void hookOnComplete() {
                        System.out.println("hookOnComplete");
                    }

                    @Override
                    protected void hookOnError(Throwable throwable) {
                        System.out.println("hookOnError");
                    }

                    @Override
                    protected void hookOnCancel() {
                        System.out.println("hookOnCancel");
                    }

                    @Override
                    public void hookOnSubscribe(Subscription subscription) {
                        request(5);
                    }

                    @Override
                    public void hookOnNext(Integer integer) {
                        System.out.println("hookOnNext");
                        System.out.println(integer);
                    }
                });


    }
}
